package cn.topcodes.tcsf.admin.jms;

import cn.topcode.unicorn.utils.PropertyConfigurer;
import cn.topcode.unicorn.utils.StringUtil;
import cn.topcode.unicorn.utils.dist.DestType;
import cn.topcode.unicorn.utils.dist.DistEventConfig;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * Created by Unicorn on 2018/5/2.
 */
public class DistEventRegister implements ApplicationContextAware {


    /**
     * 消费的事件
     */
    private Map<String, Class> consume = new HashMap<>();

    /**
     * 生产的事件
     */
    private Map<String, Class> produce = new HashMap<>();

    /**
     * 监听的Queue
     */
    private Map<String, Integer> queueDests = new HashMap<>();

    /**
     * 监听的Topic
     */
    private Map<String, Integer> topicDests = new HashMap<>();

    /**
     * 监听的VirtualTopic
     */
    private Map<String, Integer> virtualTopicDests = new HashMap<>();


    private List<Class> mConsume;

    private List<Class> mProduce;

    @Autowired(required = false)
    private PropertyConfigurer propertyConfigurer;

    private ApplicationContext applicationContext;

    public DistEventRegister(List<Class> consume, List<Class> produce) {
        if(consume.isEmpty() && produce.isEmpty()) {
            throw new IllegalArgumentException("请至少设置一个生产或消费的分布式事件类型");
        }

        this.mConsume = consume;
        this.mProduce = produce;
    }

    @PostConstruct
    public void init() {

        for(Class clazz : mProduce) {
            DistEventConfig config = (DistEventConfig) clazz.getAnnotation(DistEventConfig.class);
            if(config != null) {
                if(StringUtil.isNotBlank(config.type())) {
                    this.produce.put(config.type(), clazz);
                }else {
                    this.produce.put(clazz.getName(), clazz);
                }
            }
        }

        String consumerName = null;
        for(Class clazz : mConsume) {
            DistEventConfig config = (DistEventConfig) clazz.getAnnotation(DistEventConfig.class);
            if(config != null) {
                if(StringUtil.isNotBlank(config.type())) {
                    this.consume.put(config.type(), clazz);
                }else {
                    this.consume.put(clazz.getName(), clazz);
                }

                String destName = config.destName();
                if(StringUtil.isBlank(destName)) {
                    destName = clazz.getName();
                }
                if(DestType.QUEUE.equals(config.destType())) {
                    if(queueDests.containsKey(destName)) {
                        Integer originConcurrentConsumers = queueDests.get(destName);
                        if(config.concurrentConsumers() > originConcurrentConsumers.intValue()) {
                            queueDests.put(destName, config.concurrentConsumers());
                        }
                    }else {
                        queueDests.put(destName, config.concurrentConsumers());
                    }
                }else if(DestType.TOPIC.equals(config.destType())) {
                    topicDests.put(destName, config.concurrentConsumers());
                }else if(DestType.VIRTUAL_TOPIC.equals(config.destType())) {
                    if(StringUtil.isBlank(consumerName)) {
                        consumerName = propertyConfigurer.getProperty("activemq.VirtualTopic.consumer.name");
                        if(StringUtil.isBlank(consumerName)) {
                            throw new IllegalArgumentException("请设置activemq.VirtualTopic.consumer.name");
                        }
                    }

                    String key = "Consumer." + consumerName + ".VirtualTopic." + destName;
                    if(virtualTopicDests.containsKey(key)) {
                        Integer originConcurrentConsumers = virtualTopicDests.get(key);
                        if(config.concurrentConsumers() > originConcurrentConsumers.intValue()) {
                            virtualTopicDests.put(key, config.concurrentConsumers());
                        }
                    }else {
                        virtualTopicDests.put(key, config.concurrentConsumers());
                    }
                }
            }
        }

        mConsume.retainAll(mProduce);
        if(!mConsume.isEmpty()) {
            throw new IllegalArgumentException("生产和消费的分布式事件类型具有交集，这将产生死循环");
        }


        regist();
    }

    /**
     * 判断本程序是否生产此分布式事件类型
     * @param type
     * @return
     */
    public boolean exist(String type) {
        return produce.containsKey(type);
    }

    public Class get(String type) {
        return consume.get(type);
    }

    private void regist() {
        Map<String, Integer> destQueues = new HashMap<>();
        destQueues.putAll(queueDests);
        destQueues.putAll(virtualTopicDests);

        if(destQueues.isEmpty() && topicDests.isEmpty()) {
            return;
        }

        ConfigurableApplicationContext applicationContext = (ConfigurableApplicationContext) this.applicationContext;
        DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();

        //  注册消息处理器
        BeanDefinitionBuilder distMessageReceiverBuilder = BeanDefinitionBuilder.genericBeanDefinition(DistMessageReceiver.class);
        distMessageReceiverBuilder.setLazyInit(false);
        defaultListableBeanFactory.registerBeanDefinition("distMessageReceiver", distMessageReceiverBuilder.getRawBeanDefinition());

        //  注册监听队列bean
        for(Map.Entry<String, Integer> dest : destQueues.entrySet()) {
            String destName = dest.getKey();
            Integer concurrentConsumers = dest.getValue();

            //  构建队列目的地
            BeanDefinitionBuilder queueBuilder = BeanDefinitionBuilder.genericBeanDefinition(ActiveMQQueue.class);
            queueBuilder.addConstructorArgValue(destName);
            String destBeanName = destName.replace(".", "") + "Queue";
            defaultListableBeanFactory.registerBeanDefinition(
                    destBeanName,
                    queueBuilder.getRawBeanDefinition());

            registBean(destName, destBeanName, concurrentConsumers, defaultListableBeanFactory);
        }

        //  注册订阅主题bean
        for(Map.Entry<String, Integer> dest : topicDests.entrySet()) {
            String destName = dest.getKey();
            //  Topic只能用一个线程
            Integer concurrentConsumers = 1/*dest.getValue()*/;

            //  构建队列目的地
            BeanDefinitionBuilder topicBuilder = BeanDefinitionBuilder.genericBeanDefinition(ActiveMQTopic.class);
            topicBuilder.addConstructorArgValue(destName);
            String destBeanName = destName.replace(".", "") + "Topic";
            defaultListableBeanFactory.registerBeanDefinition(
                    destBeanName,
                    topicBuilder.getRawBeanDefinition());

            registBean(destName, destBeanName, concurrentConsumers, defaultListableBeanFactory);
        }

    }

    private void registBean(String destName, String destBeanName, Integer concurrentConsumers,
                        DefaultListableBeanFactory defaultListableBeanFactory) {
        BeanDefinitionBuilder containerBuilder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class);
        containerBuilder.setLazyInit(false);

        containerBuilder.addPropertyValue("concurrentConsumers", concurrentConsumers);
        containerBuilder.addPropertyReference("connectionFactory", "jmsConsumerFactory");
        containerBuilder.addPropertyReference("destination", destBeanName);
        containerBuilder.addPropertyValue("pubSubNoLocal", false);
        containerBuilder.addPropertyReference("errorHandler", "jmsErrorHandler");
        containerBuilder.addPropertyReference("messageListener", "distMessageReceiver");

        String beanName = destName.replace(".", "") + "MessageListenerContainer";
        defaultListableBeanFactory.registerBeanDefinition(beanName, containerBuilder.getRawBeanDefinition());
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
